package com.blb.edusearchservice.listener;

import com.alibaba.fastjson.JSON;
import com.blb.common.config.RabbitMQConfig;
import com.blb.common.entity.Course;
import com.blb.edusearchservice.dao.ElasticsearchDAO;
import com.blb.edusearchservice.entity.ElasticsearchEntity;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * 课程消息队列监听器
 */
@Slf4j
@Component
public class CourseMQListener {

    @Resource
    private ElasticsearchDAO elasticsearchDAO;

    @Resource
    private RedisTemplate<String,String> redisTemplate;

    /**
     * 监听保存课程队列
     */
    @RabbitListener(queues = {RabbitMQConfig.COURSE_SAVE_QUEUE})
    public void handleSaveCourseMessage(String json, Channel channel, Message message){
        log.info("接收到保存课程消息：{}",json);
        try {
            //处理消息的幂等性，获得消息的id
            String id = message.getMessageProperties().getHeader("spring_returned_message_correlation");
            //执行redis的setNX命令，如果不存在就保存0，执行完业务后设置为1
            ValueOperations<String, String> ops = redisTemplate.opsForValue();
            if(ops.setIfAbsent(id,"0",100, TimeUnit.SECONDS)){
                log.info("消息第一次发送{}",id);
                Course course = JSON.parseObject(json,Course.class);
                ElasticsearchEntity entity = new ElasticsearchEntity(String.valueOf(course.getId()),course);
                elasticsearchDAO.saveOrUpdate("course",entity);
                //执行完业务后，手动确认消息 1.消息id 2.批量确认
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                //执行完业务后，修改消息的状态为1
                ops.set(id,"1");
            }else{
                log.info("消息重复发送，放弃执行{}",id);
                if("1".equals(ops.get(id))){
                    log.info("消息{}已经执行完毕",id);
                }
            }
        } catch (Exception e) {
            log.error("保存课程出现异常",e);
        }
    }

    /**
     * 监听删除课程队列
     */
    @RabbitListener(queues = {RabbitMQConfig.COURSE_REMOVE_QUEUE})
    public void handleRemoveCourseMessage(Long id, Channel channel, Message message){
        log.info("接收到删除课程消息：{}",id);
        try {
            elasticsearchDAO.removeById("course",String.valueOf(id));
            //执行完业务后，手动确认消息 1.消息id 2.批量确认
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        } catch (Exception e) {
            log.error("删除课程出现异常",e);
        }
    }
}
